package com.dc.dbus.transport.netty.client;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import com.dc.dbus.transport.commons.ClientSideReceiver;
import com.dc.dbus.transport.commons.ClientSideTransport;
import com.dc.dbus.transport.commons.ServiceChangeListener;
import com.dc.dbus.transport.netty.Log;
import com.dc.dbus.transport.netty.Msg;
import com.dc.dbus.transport.netty.client.exception.CantFindServiceException;
import com.dc.dbus.transport.netty.client.node.NodeInfo;
import com.dc.dbus.transport.netty.client.node.NodeLink;

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

public class NettyClientSideTransport implements ClientSideTransport {
	
	protected final int nodeChannelNum;
	
	protected ClientSideReceiver clientSideReceiver;
	
	protected final ConcurrentHashMap<String, Service> serviceNameMapService = new ConcurrentHashMap<>();
	
	protected final ConcurrentHashMap<String, NodeLink> nodeNameMapClientLink = new ConcurrentHashMap<>();
	
	protected final ConcurrentHashMap<NodeLink, NodeLink> needCleanNode = new ConcurrentHashMap<>();
	
	protected final ServiceChangeListener serviceChangeListener = new ServiceChangeListener() {
		@Override
		public void serviceDisconected(String serviceName, String nodeName) {
			
			Log.nettyLinkLogger.error("NettyClientSideTransport.serviceChangeListener.serviceDisconected: node disconected, serviceName:" + serviceName + ", nodeName" + nodeName);
			
			Service service = serviceNameMapService.get(serviceName);
			if (service != null) {
				service.removeClient(nodeName);
			}
			nodeNameMapClientLink.remove(nodeName);
		}
		@Override
		public void serviceActive(String serviceName, String node, String[] routers, String[] hosts, int[] ports) {
			
			Log.nettyLinkLogger.info("NettyClientSideTransport.serviceChangeListener.serviceActive: node active, serviceName:" + serviceName + ", nodeName" + node
					 + ", routers"+ Arrays.toString(routers)
					 + ", ips"+ Arrays.toString(hosts)
					 + ", ports"+ Arrays.toString(ports));
			
			Service service = serviceNameMapService.get(serviceName);
			if (service == null) {
				service = new Service(NettyClientSideTransport.this, serviceName);
				Service old = serviceNameMapService.putIfAbsent(serviceName, service);
				if (old !=null) {
					service = old;
				}
			}
			
			service.addClient(node, routers, hosts, ports);
		}
		
		@Override
		public void serviceChange(String serviceName, String node, String[] routers, String[] hosts, int[] ports) {
			
			Log.nettyLinkLogger.info("NettyClientSideTransport.serviceChangeListener.serviceChange: node change, serviceName:" + serviceName + ", nodeName" + node
					 + ", routers"+ Arrays.toString(routers)
					 + ", ips"+ Arrays.toString(hosts)
					 + ", ports"+ Arrays.toString(ports));
			
			Service service = serviceNameMapService.get(serviceName);
			if (service == null) {
				service = new Service(NettyClientSideTransport.this, serviceName);
				Service old = serviceNameMapService.putIfAbsent(serviceName, service);
				if (old !=null) {
					service = old;
				}
			}
			
			service.clientChange(node, routers, hosts, ports);
		}
	};
	
	protected final ClientChannelDisconnectHandler channelDisconnectHandler = new ClientChannelDisconnectHandler() {
		@Override
		public void channelDisconnect(NettyClient nettyClient) {
			NettyClientSideTransport.this.channelDisconnect(nettyClient);
		}
	};
	
	protected final Thread tryLinkFailNodeThread;
	
	protected final EventLoopGroup nettyEventLoopGroup = new NioEventLoopGroup();
	
	public NettyClientSideTransport(int nodeChannelNum) {
		
		this.nodeChannelNum = nodeChannelNum;
		
		this.tryLinkFailNodeThread = new Thread(new Runnable() {
			@Override
			public void run() {
				tryLinkFailNodeAndCleanChannel();
			}
		});
		this.tryLinkFailNodeThread.setName("TryLinkFailNodeThread");
		this.tryLinkFailNodeThread.setDaemon(true);
		this.tryLinkFailNodeThread.start();
	}
	
	public ServiceChangeListener getServiceChangeListener() {
		return serviceChangeListener;
	}
	
	@Override
	public void startClientSideReceiver(ClientSideReceiver clientSideReceiver) {
		
		this.clientSideReceiver = clientSideReceiver;
	}
	
	@Override
	public void send(String serviceName, String router, long msgId, int cmd, Object param, int invokeSessionTimeout) {
		
		if (Log.nettyLinkLogger.isDebugEnabled()) {
			Log.nettyLinkLogger.debug("NettyClientSideTransport.send: send to server:" + serviceName
					+ ", router:" + router + ", msgId" + msgId + ", invokeSessionTimeout:" + invokeSessionTimeout
					+ ", param:" + param);
		}
		
		Service service = serviceNameMapService.get(serviceName);
		if (service == null) {
			throw new CantFindServiceException("unknow service:" + serviceName);
		}
		
		Msg msg = new Msg(msgId, cmd, param);
		service.writeAndFlush(router, msg, invokeSessionTimeout);
	}
	
	protected NodeLink getNodeLink(String nodeName) {
		
		NodeLink oldNode = nodeNameMapClientLink.get(nodeName);
		return oldNode;
	}
	
	protected NodeLink getOrCreateNodeLink(String nodeName, NodeInfo node) {
		
		NodeLink oldNode = nodeNameMapClientLink.get(nodeName);
		if (oldNode != null) {
			return oldNode;
		} else {
			
			NodeLink newNode = newNodeLInk(node);
			oldNode = nodeNameMapClientLink.putIfAbsent(nodeName, newNode);
			if (oldNode != null) {
				newNode.close();
				return oldNode;
			} else {
				return newNode;
			}
		}
	}
	
	protected void needCleanNode(NodeLink node) {
		needCleanNode.put(node, node);
	}
	
	protected NodeLink newNodeLInk(NodeInfo node) {

		Log.nettyLinkLogger.info("NettyClientSideTransport.newNodeLInk BEGIN: node:" + node);

		String serviceName = node.serviceName;
		String nodeName = node.nodeName;
		String[] routers = node.routers;
		String[] hosts = node.hosts;
		int[] ports = node.ports;
		
		ArrayList<NettyClient> clients = new ArrayList<>();
		for (int channelCount = 0, i=0; channelCount < nodeChannelNum && i<hosts.length; ) {
			try {
				NettyClient c = new NettyClient(nettyEventLoopGroup, nodeName, hosts[i], ports[i],
						clientSideReceiver, channelDisconnectHandler);
				clients.add(c);

				channelCount++;

			} catch (Exception e) {
				Log.nettyLinkLogger.info("NettyClientSideTransport.newNodeLInk: can't link to " + nodeName + "(" + hosts[i] + ":" + ports[i] + ")", e);
				i++;
			}
		}
		
		NettyClient[] clientArray = clients.toArray(new NettyClient[clients.size()]);
		NodeLink nodeLink = new NodeLink(serviceName, nodeName, routers, hosts, ports, clientArray);

		Log.nettyLinkLogger.info("NettyClientSideTransport.newNodeLInk END: node:" + node + ", nodeLink:" + nodeLink);

		return nodeLink;
	}
	
	public void channelDisconnect(NettyClient nettyClient) {
		
		Log.nettyLinkLogger.warn("NettyClientSideTransport.channelDisconnect: channel distconnect:" + nettyClient.getHost() + ":" + nettyClient.getPort());
		
		NodeLink node = nodeNameMapClientLink.get(nettyClient.getNodeName());
		
		if (node == null) {
			return;
		}
		
		synchronized (node) {
			
			ArrayList<NettyClient> clients = new ArrayList<>();
			
			NettyClient[] clientArray = node.clients;
			
			for (int i=0; i<clientArray.length; i++) {
				NettyClient c = clientArray[i];
				if (c != nettyClient) {
					clients.add(clientArray[i]);
				}
			}
			
			NettyClient[] newClientArray = clients.toArray(new NettyClient[clients.size()]);
			
			node.clients = newClientArray;
			
		}
	}
	
	protected void tryLinkFailNodeAndCleanChannel() {
		
		for (;;) {
			
			try {
				
				boolean hasFail = false;
				
				for (NodeLink n : nodeNameMapClientLink.values()) {
					
					if (n.clients.length < nodeChannelNum) {
						
						Log.nettyLinkLogger.warn("NettyClientSideTransport.tryLinkFailNodeAndCleanChannel: try reconnect to node:" + n);
						synchronized (n) {
							ArrayList<NettyClient> clients = new ArrayList<>();
							
							NettyClient[] clientArray = n.clients;
							
							String nodeName = n.nodeName;
							String[] hosts = n.hosts;
							int[] ports = n.ports;
							
							for (int i=0; i<clientArray.length; i++) {
								clients.add(clientArray[i]);
							}
							
							for (int channelCount = clientArray.length, i=0; channelCount < nodeChannelNum && i< hosts.length; ) {

								try {
									NettyClient c = new NettyClient(nettyEventLoopGroup, nodeName, hosts[i], ports[i],
											clientSideReceiver, channelDisconnectHandler);
									clients.add(c);
									channelCount++;
									Log.nettyLinkLogger.info("NettyClientSideTransport.tryLinkFailNodeAndCleanChannel: connect to node:" + n.nodeName
											+ " SUCCESS, host:" + hosts[i] + ":" + ports[i]);
								} catch (Exception e) {
									Log.nettyLinkLogger.info("NettyClientSideTransport.tryLinkFailNodeAndCleanChannel: can't link to " + nodeName + "(" + hosts[i] + ":" + ports[i] + ")", e);
									i++;
								}
							}
							
							NettyClient[] newClientArray = clients.toArray(new NettyClient[clients.size()]);
							
							if (newClientArray.length < nodeChannelNum) {
								hasFail = true;
							}
							
							n.clients = newClientArray;
						}
					}
				}
				
				long timeNow = System.currentTimeMillis();
				for (NodeLink n : needCleanNode.values()) {
					if (n.getInvokeSessionExpireTime() < timeNow) {
						needCleanNode.remove(n);
						try {
							n.close();
						} catch (Exception e) {
							Log.nettyLinkLogger.error("NodeLink.close Exception:", e);
						}
						Log.nettyLinkLogger.info("NettyClientSideTransport.tryLinkFailNodeAndCleanChannel: close not useful node:" + n);
					}
				}
				
				if (hasFail) {
					TimeUnit.MILLISECONDS.sleep(50);
				} else {
					TimeUnit.MILLISECONDS.sleep(500);
				}
				
			} catch (Exception e) {
				Log.nettyLinkLogger.error("NettyClientSideTransport.tryLinkFailNodeAndCleanChannel: Exception:", e);

				try {
					TimeUnit.MILLISECONDS.sleep(50);
				} catch (InterruptedException ex) {
					Log.nettyLinkLogger.error("NettyClientSideTransport.tryLinkFailNodeAndCleanChannel: Exception:", ex);
				}
			}
			
		}
	}
	
}
